package com.meiya.hugegraph.mock;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.node.ObjectNode;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.math.BigDecimal;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.*;
import java.util.concurrent.TimeUnit;

/**
 * 模拟交易数据生成器
 * 根据真实数据格式生成模拟数据并推送到Kafka
 */
public class TransactionDataGenerator {
    private static final Logger logger = LoggerFactory.getLogger(TransactionDataGenerator.class);
    private static final String BOOTSTRAP_SERVERS = "192.168.182.128:9092";
    private static final String TOPIC = "transaction-data";
    private static final int NUM_RECORDS = 1000;
    // 每条记录生成间隔时间 ms
    private static final int DELAY_MS = 1;

    private static final Random random = new Random();
    private static final ObjectMapper objectMapper = new ObjectMapper();
    private static final DateTimeFormatter dateFormatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");

    // 模拟数据的基础信息
    private static final String[] TRANSACTION_TYPES = { "转账", "消费", "还款" };
    private static final String[] COUNTER_PARTIES = {
            "财付通-微信转账",
            "支付宝（中国）网络技术有限公司",
            "财付通-微信面对面收款",
            "财付通-微信红包",
            "支付宝-还款"
    };

    private static final Map<String, AccountDataGenerator.AccountInfo> accountInfoMap = AccountDataGenerator.getAccountInfoMap();

    // 当前批次的任务记录ID，默认使用AccountDataGenerator中的批次ID
    private static String batchTaskRecordId = AccountDataGenerator.getBatchTaskRecordId();

    /**
     * 生成交易记录
     * @return 生成的交易记录JSON字符串
     */
    private static String generateTransactionRecord() {
        return generateTransactionRecord(accountInfoMap);
    }

    private static String generateTransactionRecord(Map<String, AccountDataGenerator.AccountInfo> useAccountMap) {
        try {
            ObjectNode record = objectMapper.createObjectNode();

            // 从账户映射中随机选择一个账户
            String[] bankCards = useAccountMap.keySet().toArray(new String[0]);
            String bankCard = bankCards[random.nextInt(bankCards.length)];
            AccountDataGenerator.AccountInfo accountInfo = useAccountMap.get(bankCard);

            // 生成基本信息
            String idNumber = accountInfo.idNumber; // 使用账户中的证件号码
            LocalDateTime transTime = generateRandomTime();
            BigDecimal amount = generateAmount();
            BigDecimal balance = generateBalance();
            String direction = amount.compareTo(BigDecimal.ZERO) > 0 ? "进" : "出";
            
            // 生成对手账号，有10%的概率使用已有的银行卡作为对手账号
            String counterPartyAccount;
            String counterPartyName;
            String counterPartyIdNumber = null;
            
            if (random.nextInt(10) == 0 && bankCards.length > 1) {
                // 从已有账户中选择一个不同的账户作为对手
                String otherBankCard;
                do {
                    otherBankCard = bankCards[random.nextInt(bankCards.length)];
                } while (otherBankCard.equals(bankCard));
                
                counterPartyAccount = otherBankCard;
                AccountDataGenerator.AccountInfo counterPartyInfo = useAccountMap.get(otherBankCard);
                counterPartyName = counterPartyInfo.accountName;
                counterPartyIdNumber = counterPartyInfo.idNumber;
            } else {
                // 生成随机对手账号和名称
                counterPartyAccount = generateCounterPartyAccount();
                counterPartyName = COUNTER_PARTIES[random.nextInt(COUNTER_PARTIES.length)];
            }
            
            String transactionType = TRANSACTION_TYPES[random.nextInt(TRANSACTION_TYPES.length)];
            long ukeyValue = random.nextLong();
            String currentTime = LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));
            
            // 设置交易记录字段 - 按照需求中的transaction字段规范填充
            record.put("jykh", bankCard); // 交易卡号
            record.put("jyzh", bankCard); // 交易账号
            record.put("jyfhm", accountInfo.accountName); // 交易方户名
            record.put("jyfzjhm", idNumber); // 交易方证件号码
            record.put("jysj", transTime.format(dateFormatter)); // 交易时间
            record.put("jyje", amount.abs().toString()); // 交易金额
            record.put("jyye", balance.toString()); // 交易余额
            record.put("sfbz", direction); // 收付标志（出/进）
            record.put("jydszkh", counterPartyAccount); // 交易对手账号
            record.putNull("xjbz"); // 现金标志
            record.put("dshm", counterPartyName); // 对手户名
            
            // 如果对手是已有账户，则设置对手身份证号
            if (counterPartyIdNumber != null) {
                record.put("dssfzh", counterPartyIdNumber); // 对手身份证号
            } else {
                record.putNull("dssfzh"); // 对手身份证号
            }
            
            record.putNull("dskhyh"); // 对手开户银行
            record.put("zysm", generateTransactionDesc()); // 摘要说明
            record.put("jybz", "CNY"); // 交易币种
            record.put("jywdmc", "深圳:建行深圳市分行资金结算与营运管理部"); // 交易网点名称
            record.putNull("jyfsd"); // 交易发生地
            record.put("jysfcg", "1"); // 交易是否成功
            record.putNull("cph"); // 车牌号
            record.putNull("ipdz"); // IP地址
            record.putNull("macdz"); // MAC地址
            record.putNull("dsjyye"); // 对手交易余额
            record.put("jylsh", generateTransactionId()); // 交易流水号
            record.putNull("rzh"); // 入账号
            record.putNull("pzzl"); // 凭证种类
            record.putNull("pzh"); // 凭证号
            record.put("jygyh", "00000000"); // 交易柜员号
            record.put("bz", counterPartyName); // 备注
            record.putNull("shmc"); // 商户名称
            record.put("jylx", transactionType); // 交易类型
            record.put("cxfkjgyy", "成功"); // 查询反馈结果原因
            record.put("clue_id", AccountDataGenerator.getBatchClueId()); // 使用批次级别的clue_id
            record.put("task_record_id", batchTaskRecordId); // 使用批次任务记录ID
            record.put("a_yhlx", accountInfo.bankName); // 银行类型
            record.put("action_type", "0"); // 操作类型（0 原始数据，1 反向数据）
            record.put("from_jykh", direction.equals("进") ? counterPartyAccount : bankCard); // 来源交易卡号
            record.put("to_jydszkh", direction.equals("进") ? bankCard : counterPartyAccount); // 目标交易对手账号
            record.put("crrq", currentTime); // 插入日期
            record.put("ukey", ukeyValue); // UKey(计算Hash后的唯一值)

            return objectMapper.writeValueAsString(record);
        } catch (Exception e) {
            throw new RuntimeException("生成交易记录时出错", e);
        }
    }

    private static LocalDateTime generateRandomTime() {
        // 生成最近7天内的随机时间
        return LocalDateTime.now().minusDays(random.nextInt(7))
                .minusHours(random.nextInt(24))
                .minusMinutes(random.nextInt(60));
    }

    private static BigDecimal generateAmount() {
        // 生成-10000到10000之间的随机金额
        return new BigDecimal(random.nextDouble() * 20000 - 10000)
                .setScale(2, BigDecimal.ROUND_HALF_UP);
    }

    private static BigDecimal generateBalance() {
        // 生成0到100000之间的随机余额
        return new BigDecimal(random.nextDouble() * 100000)
                .setScale(2, BigDecimal.ROUND_HALF_UP);
    }

    private static String generateCounterPartyAccount() {
        StringBuilder account = new StringBuilder("10");
        for (int i = 0; i < 10; i++) {
            account.append(random.nextInt(10));
        }
        return account.toString();
    }

    private static String generateTransactionDesc() {
        String[] descs = { "消费", "转账", "还款", "充值", "提现" };
        return descs[random.nextInt(descs.length)];
    }

    private static String generateTransactionId() {
        return "101001" + random.nextInt(100) + System.currentTimeMillis();
    }

    /**
     * 主方法，用于独立运行测试
     */
    public static void main(String[] args) {
        // 如果是独立运行，则生成新的批次任务记录ID
        if (AccountDataGenerator.getBatchTaskRecordId() == null) {
            batchTaskRecordId = UUID.randomUUID().toString();
        } else {
            batchTaskRecordId = AccountDataGenerator.getBatchTaskRecordId();
        }
        logger.info("使用批次任务记录ID: {}", batchTaskRecordId);
        
        Properties props = new Properties();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP_SERVERS);
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

        try (KafkaProducer<String, String> producer = new KafkaProducer<>(props)) {
            logger.info("开始生成并发送交易数据...");

            for (int i = 0; i < NUM_RECORDS; i++) {
                String record = generateTransactionRecord();
                producer.send(new ProducerRecord<>(TOPIC, record));

                if (i % 100 == 0) {
                    logger.info("已发送 {} 条记录", i);
                }

                if (DELAY_MS > 0) {
                    TimeUnit.MILLISECONDS.sleep(DELAY_MS);
                }
            }

            logger.info("数据生成完成，共发送 {} 条记录", NUM_RECORDS);
        } catch (Exception e) {
            logger.error("发送数据时出错: {}", e.getMessage(), e);
        }
    }
}